package com.gin.monitor.agg.warn;

import com.gin.monitor.agg.connector.kafka.FlinkKafkaUtils;
import com.gin.monitor.agg.utils.JobEnvUtils;
import com.gin.monitor.agg.vo.RepetitionCarWarning;
import com.gin.monitor.agg.vo.TrafficInfo;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;


/**
 * 套牌车分析，从数据流中找出存在套牌嫌疑的车辆
 * <p>
 * cd /opt/software/kafka_2.11-2.0.0/bin
 * bin/kafka-server-start.sh -daemon config/server.properties
 * ./kafka-console-consumer.sh --bootstrap-server node01:9092 --topic t_traffic_monitor --group group01 --property print.key=true --property print.value=true --property key.separator=,
 * 测试生产
 * ./kafka-console-producer.sh --broker-list node01:9092 --topic t_traffic_monitor
 * <p>
 * 测试时注意: 使用当前时间戳
 * 1620637883000,0002,75312,京L22188,60.1,27,31
 * 1620637884000,0002,75312,京L22188,70.1,27,31
 * 1620637885000,0002,75312,京L22188,80.1,27,31
 * 1620637885000,0002,75312,京L22188,80.1,27,31
 *
 * @author gin
 * @date 2021/4/30
 */
public class RepetitionCarAnalysis {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = JobEnvUtils.initEnv(args);

        //streamKafka 海量的数据流，不可以存入广播状态流中
        //streamMysql 从Mysql数据库中读取的卡口限速信息，特点：数据量少，更新不频繁

        //默认最近开始读取
        //DataStreamSource<TrafficInfo> streamKafka = env.addSource(FlinkKafkaUtils.getKafkaConsumer());
        //从第一行开始读取. 测试用!
        SingleOutputStreamOperator<TrafficInfo> streamKafka = env
                .addSource(FlinkKafkaUtils.getKafkaConsumer().setStartFromEarliest())
                //指定时间戳字段, 及允许延时时间
                .assignTimestampsAndWatermarks(WatermarkStrategy
                        .<TrafficInfo>forBoundedOutOfOrderness(Duration.ofSeconds(20))
                        .withTimestampAssigner((event, timestamp) -> event.actionTime));

        //streamKafka.print();

        streamKafka
                .keyBy(new KeySelector<TrafficInfo, String>() {
                    @Override
                    public String getKey(TrafficInfo trafficInfo) throws Exception {
                        return trafficInfo.car;
                    }
                })
                .process(new KeyedProcessFunction<String, TrafficInfo, RepetitionCarWarning>() {

                    ValueState<TrafficInfo> firstState;

                    @Override
                    public void open(Configuration parameters) {
                        //状态中保存第一辆出现的信息对象
                        firstState = getRuntimeContext().getState(new ValueStateDescriptor<TrafficInfo>("first", TrafficInfo.class));
                    }

                    @Override
                    public void processElement(TrafficInfo value, Context ctx, Collector<RepetitionCarWarning> out) throws Exception {
                        TrafficInfo first = firstState.value();
                        if (first == null) { //当前这量车就是第一辆车
                            firstState.update(value);
                        } else {
                            //有两辆车出现了，但是时间已经超过了10秒，另外一种情况是时间没有超过10秒(涉嫌套牌)
                            Long nowTime = value.actionTime;
                            Long firstTime = first.actionTime;
                            //可能乱序, 取绝对值
                            Long less = Math.abs(nowTime - firstTime) / 1000;
                            if (less <= 10) { //涉嫌
                                RepetitionCarWarning warn = new RepetitionCarWarning(value.car,
                                        nowTime > firstTime ? first.monitorId : value.monitorId,
                                        nowTime <= firstTime ? first.monitorId : value.monitorId,
                                        "涉嫌套牌车",
                                        ctx.timerService().currentProcessingTime()
                                );
                                out.collect(warn);
                                firstState.clear();
                            } else {
                                //不是套牌车，把第二次经过卡口的数据保存到状态中，以便下次判断
                                if (nowTime > firstTime) {
                                    firstState.update(value);
                                }
                            }
                        }
                    }

                })
        .print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

}
